package com.example.springbootbinlog.consumer;

import com.example.springbootbinlog.config.DatabaseConfig;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.Serializable;
import java.util.*;

@Component
@Slf4j
public class BinlogConsumer implements CommandLineRunner {

    @Autowired
    DatabaseConfig databaseConfig;


    @Override
    public void run(String... args) {

        log.info("binlog listener start...");
        List<String> databaseList = Arrays.asList(databaseConfig.getTableName().split(","));
        HashMap<Long, String> tableMap = new HashMap<>();
        BinaryLogClient binaryLogClient = new BinaryLogClient(databaseConfig.getHost(), databaseConfig.getPort(), databaseConfig.getUser(), databaseConfig.getPassword());
        binaryLogClient.setServerId(databaseConfig.getServerId());
        binaryLogClient.registerEventListener((event -> {
            // binlog事件
            EventData data = event.getData();
            if (data != null) {
//                System.out.println(data.getClass());
                if (data instanceof TableMapEventData) {
                    TableMapEventData tableMapEventData = (TableMapEventData) data;
                    tableMap.put(tableMapEventData.getTableId(), tableMapEventData.getDatabase() + "." + tableMapEventData.getTable());
                }
                // update数据
                if (data instanceof UpdateRowsEventData) {
                    UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
                    String tableName = tableMap.get(updateRowsEventData.getTableId());
                    if (tableName != null && databaseList.contains(tableName)) {
                        String eventKey = tableName + ".update";
                        for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
                            List<String> value = new ArrayList<>();
                            for (Serializable s : row.getValue()) {
                                value.add(s.toString());
                            }
                            System.out.println(eventKey + "value = " + value.toString());

                        }
                    }
                }
                // insert数据
                else if (data instanceof WriteRowsEventData) {
                    WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
                    String tableName = tableMap.get(writeRowsEventData.getTableId());
                    if (tableName != null && databaseList.contains(tableName)) {
                        String eventKey = tableName + ".insert";
                        for (Serializable[] row : writeRowsEventData.getRows()) {
                            List<String> value = new ArrayList<>();
                            for (Serializable s : row) {
                                value.add(s.toString());
                            }
                            System.out.println(eventKey + "value = " + value.toString());
                        }
                    }
                }
                // delete数据
                else if (data instanceof DeleteRowsEventData) {
                    DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
                    String tableName = tableMap.get(deleteRowsEventData.getTableId());
                    if (tableName != null && databaseList.contains(tableName)) {
                        String eventKey = tableName + ".delete";
                        for (Serializable[] row : deleteRowsEventData.getRows()) {
                            List<String> value = new ArrayList<>();
                            for (Serializable s : row) {
                                value.add(s.toString());
                            }
                            System.out.println(eventKey + "value = " + value.toString());
                        }
                    }
                }
            }
        }));
        try {
            binaryLogClient.connect();
            log.info("Listen on the database successfully");
        } catch (IOException e) {
            log.error("db connection error cause by {}", e.getMessage(), e);
        }
    }
}
